﻿using LJ.EventBus.HornedSerpent.Transport.Abstractions;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace LJ.EventBus.HornedSerpent.Transport.InMemoryQueue.Internal;

public class InMemoryEventBusQueueConsumerClient(IMemoryQueue memoryQueue) : IEventBusConsumerClient, IInMemoryQueueConsumerClient
{
    private readonly IMemoryQueue _memoryQueue = memoryQueue;
    private string _group = null!;
    private string _topic = null!;
    private bool disposedValue;
    private readonly BlockingCollection<EventBusTransportData> _transportDataQueue = [];

    public void Init(string group, string topic)
    {
        _group = group;
        _topic = topic;
        _memoryQueue.RegisterConsumerClient(group, this);
    }
    public Func<EventBusTransportData, object, Task>? OnDataCallback { get; set; }
    public Action<EventBusTransportLog>? OnLogCallback { get; set; }

    public void FetchTopic()
    {
        _memoryQueue.Subscribe(_group, [_topic]);
    }

    public Task ListeningAsync(TimeSpan timeout, CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();
        foreach (var data in _transportDataQueue.GetConsumingEnumerable(cancellationToken))
        {
            OnDataCallback?.Invoke(data, _transportDataQueue);
        }

        return Task.CompletedTask;
    }

    public void Commit(object? sender)
    {
    }

    public void Reject(object? sender)
    {
    }

    public void Subscribe()
    {
        _memoryQueue.Subscribe(_group, [_topic]);
    }

    public void AddSubscribeData(EventBusTransportData transportData)
    {
        _transportDataQueue.Add(transportData);
    }

    #region Dispose
    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                // TODO: 释放托管状态(托管对象)
                _memoryQueue.Unsubscribe(_group);
            }

            // TODO: 释放未托管的资源(未托管的对象)并重写终结器
            // TODO: 将大型字段设置为 null
            disposedValue = true;
        }
    }

    // // TODO: 仅当“Dispose(bool disposing)”拥有用于释放未托管资源的代码时才替代终结器
    // ~InMemoryQueueConsumerClient()
    // {
    //     // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
    //     Dispose(disposing: false);
    // }

    public void Dispose()
    {
        // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }

   
    #endregion
}
